package com.learn.serialier;

import com.learn.bean.ToKafkaSinkPojo;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.kafka.clients.producer.ProducerRecord;

import javax.annotation.Nullable;
import java.nio.charset.StandardCharsets;

public class SerializationAian implements KafkaSerializationSchema<ToKafkaSinkPojo> {

    String topic;

    public SerializationAian() {
    }

    public SerializationAian(String topic) {
        this.topic = topic;
    }


    @Override
    public ProducerRecord<byte[], byte[]> serialize(ToKafkaSinkPojo element, @Nullable Long timestamp) {
        String key = element.getKey();

        String value = element.getValue();

        ProducerRecord producerRecord = new ProducerRecord<byte[], byte[]>(topic, null, null, key.getBytes(StandardCharsets.UTF_8), value.getBytes(StandardCharsets.UTF_8));

        return producerRecord;
    }
}
